package org.lqk.file.transfer.client.netty;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.lqk.file.transfer.CommandCallBack;
import org.lqk.file.transfer.RemotingClient;
import org.lqk.file.transfer.client.framework.FileResponseProcessorDispatcher;
import org.lqk.file.transfer.netty.RequestFuture;
import org.lqk.file.transfer.protocol.CommandBody;
import org.lqk.file.transfer.protocol.FileSegmentRequest;
import org.lqk.file.transfer.protocol.FileSegmentResponse;
import org.lqk.file.transfer.protocol.RemotingCommand;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.Iterator;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Created by Administrator on 2015/8/25.
 */
public class NettyRemotingClient implements RemotingClient {

    private final Bootstrap bootstrap;

    private final EventLoopGroup eventWorkerLoopGroup;

    private ConcurrentHashMap<String, ChannelFuture> channelMap = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<Integer, RequestFuture> requestTable = new ConcurrentHashMap<Integer, RequestFuture>(
			256);
    private NettyRemotingClientHelper nettyRemotingClientHelper;
    // 定时器
    private final Timer timer = new Timer("ClientHouseKeepingService", true);
    private FileResponseProcessorDispatcher fileResponseProcessorDispatcher;
    
    private static Logger log = LoggerFactory.getLogger(NettyRemotingClient.class);
    
    public NettyRemotingClient() {
        this.bootstrap = new Bootstrap();
        this.eventWorkerLoopGroup = new NioEventLoopGroup();
        fileResponseProcessorDispatcher = new FileResponseProcessorDispatcher(requestTable);
        nettyRemotingClientHelper = new NettyRemotingClientHelper(requestTable);
    }

    public void start() {
        this.bootstrap
                .group(eventWorkerLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(
                                new ObjectEncoder(),
                                new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                                new NettyClientHandler()
                        );
                    }

                });

        timer.scheduleAtFixedRate(new TimerTask() {
            @Override
            public void run() {
                scanForReSend();
            }
        }, 5 * 1000, 1000);
    }

    /**
     * 建立channel
     * @param addr
     * @return
     */
    public Channel createChannel(String addr) {
        Channel channel = null;
        ChannelFuture future = channelMap.get(addr);
        if(future != null) {
            channel = future.channel();
            return channel;
        }

        String[] split = addr.split(":");
        InetSocketAddress address = new InetSocketAddress(split[0], Integer.valueOf(split[1]));
        ChannelFuture sync = null;
        try {
            sync = this.bootstrap.connect(address).sync();
            channelMap.put(addr, sync);
            channel = sync.channel();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return channel;
    }

    /**
     * 关闭channel
     * @param addr
     * @param channel
     */
    public void closeChannel(String addr, Channel channel) {

    }

    public RemotingCommand invokeSync(String addr, RemotingCommand command, long timeoutMills) {
        try {
            Channel channel = createChannel(addr);
            if (channel != null && channel.isActive()) {
                RemotingCommand repsonse = nettyRemotingClientHelper.invokeSync(channel, command, timeoutMills);
                return repsonse;
            }else {
                this.closeChannel(addr, channel);
                throw new Exception(addr);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void invokeAsync(String addr, RemotingCommand command, long timeoutMills, CommandCallBack callBack) {
        try {
            Channel channel = createChannel(addr);
            if (channel != null && channel.isActive()) {
            	nettyRemotingClientHelper.invokeAsync(channel, command, timeoutMills, callBack);
            }else {
                this.closeChannel(addr, channel);
                throw new Exception(addr);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void invokeOneway(String addr, RemotingCommand command, long timeoutMills) {
        try {
            Channel channel = createChannel(addr);
            if (channel != null && channel.isActive()) {
            	nettyRemotingClientHelper.invokeOneway(channel, command, timeoutMills);
            }else {
                this.closeChannel(addr, channel);
                throw new Exception(addr);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        this.eventWorkerLoopGroup.shutdownGracefully();
    }

    class NettyClientHandler extends SimpleChannelInboundHandler<RemotingCommand>{

        @Override
        protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) throws Exception {
        	log.debug("opaque {},cmdCode {},rpctype {},isFlag {}",remotingCommand.getOpaque(),remotingCommand.getCmdCode(),remotingCommand.getRpcType(),remotingCommand.isFlag());
        	CommandBody body = remotingCommand.getBody();
        	if (body instanceof FileSegmentResponse) {
        		FileSegmentResponse f = (FileSegmentResponse) body;
				log.debug("position {},blocksize {},filename {},isBlockState {}, isFileState {}", f.getPosition(), f.getBlockSize(),
						f.getFileName(),f.isBlockState(),f.isFileState());

			}
        	fileResponseProcessorDispatcher.processCommand(channelHandlerContext, remotingCommand);
        }
    }
    
    /**
	 * 异步消息超时重发
	 */
	public void scanForReSend() {
		Collection<RequestFuture> values = requestTable.values();

		for (Iterator<RequestFuture> iter = values.iterator(); iter.hasNext();) {
			RequestFuture requestFuture = iter.next();
			if (requestFuture.isTimeout()) {

				RemotingCommand requestCommand = requestFuture.getRequestCommand();
				iter.remove();
				if (requestFuture.isReSendRequest()) {
					try {
						nettyRemotingClientHelper.invokeAsync(requestFuture.getChannel(), requestCommand, requestFuture.getTimeoutMills(),
								requestFuture.getCallBack());
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}

			}
		}
	}
}
